"""
Case Type   : 功能测试
Case Name   : 单条grant语句，授予多个用户代理权限，revoke回收部分用户权限
Create At   : 2024-11-11
Owner       : wan005
Description :
        1.创建多个普通用户
        2.授予用户1在当前数据库中的create权限
        3.grant授予多个用户代理权限;
        4.使用用户1在public中创建表;
        5.使用revoke回收用户2权限;
        6.使用用户2登录数据库建表、查看表数据;
        7.使用用户3登录数据库建表、查看表数据;
        8.使用revoke回收多个用户权限；
        9.使用用户3登录数据库建表、查看表数据；
Expect      :
        1.创建成功;
        2.授权成功
        3.授权成功
        4.创建成功
        5.权限回收成功
        6.无权限建表、查看表数据
        7.建表、查看数据成功
        8.权限回收成功
        9.无权限建表、查看数据
History     :
"""
import os
import unittest
from yat.test import macro
from testcase.utils.Common import Common
from testcase.utils.Constant import Constant
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Logger import Logger


class Compatibility(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        self.log.info(f'----{os.path.basename(__file__)} start----')
        self.pri_sh = CommonSH('PrimaryDbBmode')
        self.common = Common()
        self.constant = Constant()
        self.user1 = 'user_revoke_001'
        self.user2 = 'user_revoke_002'
        self.user3 = 'user_revoke_003'
        self.user4 = 'user_revoke_004'
        self.table1 = 'table_revoke_001'
        self.table2 = 'table_revoke_002'

    def test_revoke(self):
        text = '-----step1:创建多个普通用户;expect:创建成功-----'
        self.log.info(text)
        sql1 = f'create user {self.user1} password \'{macro.COMMON_PASSWD}\';' \
               f'create user {self.user2} password \'{macro.COMMON_PASSWD}\';' \
               f'create user {self.user3} password \'{macro.COMMON_PASSWD}\';' \
               f'create user {self.user4} password \'{macro.COMMON_PASSWD}\';'
        res1 = self.pri_sh.execut_db_sql(sql1)
        self.log.info(res1)
        self.assertEqual(4, res1.count(
            self.constant.CREATE_ROLE_SUCCESS_MSG), f'执行失败: {text}')

        text = '----step2:授予用户1在当前数据库中的create权限;expect:成功----'
        self.log.info(text)
        sql2 = f'grant create on schema public to {self.user1};'
        res2 = self.pri_sh.execut_db_sql(sql2)
        self.log.info(res2)
        self.assertEqual(self.constant.GRANT_SUCCESS_MSG, res2,
                         f'执行失败: {text}')

        text = '----step3:grant授予多个用户代理权限;expect:成功----'
        self.log.info(text)
        sql3 = f'grant proxy on {self.user1} to {self.user2}, {self.user3}, ' \
               f'{self.user4};'
        res3 = self.pri_sh.execut_db_sql(sql3)
        self.log.info(res3)
        self.assertEqual('GRANT ROLE', res3, f'执行失败: {text}')

        text = '----step4:使用用户1在public中创建表;expect:成功----'
        self.log.info(text)
        sql4 = f'create table public.{self.table1}(column_a int);'
        res4 = self.pri_sh.execut_db_sql(sql4,
                    sql_type=f'-U {self.user1} -W {macro.COMMON_PASSWD}')
        self.log.info(res4)
        self.assertEqual(self.constant.CREATE_TABLE_SUCCESS,
                         res4, f'执行失败: {text}')

        text = '----step5:使用revoke回收用户2权限;expect:成功----'
        self.log.info(text)
        sql5 = f'revoke proxy on {self.user1} from {self.user2};'
        res5 = self.pri_sh.execut_db_sql(sql5)
        self.log.info(res5)
        self.assertEqual('REVOKE ROLE', res5, f'执行失败: {text}')

        text = '----step6:使用用户2登录数据库建表、查看表数据;expect:无权限查看----'
        self.log.info(text)
        sql6 = f'create table public.{self.table2}(column_a text);' \
              f'select * from public.{self.table1};'
        res6 = self.pri_sh.execut_db_sql(sql6,
                    sql_type=f'-U {self.user2} -W {macro.COMMON_PASSWD}')
        self.log.info(res6)
        self.assertIn('permission denied for schema public', res6,
                      f'执行失败: {text}')
        self.assertIn(f'permission denied for relation {self.table1}', res6,
                      f'执行失败: {text}')

        text = '----step7:使用用户3登录数据库建表、查看表数据;expect:成功----'
        self.log.info(text)
        sql7 = f'create table public.{self.table2}(column_a text);' \
               f'select * from public.{self.table1};' \
               f'drop table public.{self.table2}'
        res7 = self.pri_sh.execut_db_sql(sql7,
                        sql_type=f'-U {self.user3} -W {macro.COMMON_PASSWD}')
        self.log.info(res7)
        self.common.equal_sql_mdg(res7, 'CREATE TABLE','column_a',
                                  '(0 rows)', '', 'DROP TABLE', flag='2')

        text = '----step8:使用revoke回收多个用户权限;expect:成功----'
        self.log.info(text)
        sql8 = f'revoke proxy on {self.user1} from {self.user3}, {self.user4};'
        res8 = self.pri_sh.execut_db_sql(sql8)
        self.log.info(res8)
        self.assertEqual('REVOKE ROLE', res8, f'执行失败: {text}')

        text = '----step9:使用用户3登录数据库建表、查看表数据;expect:失败----'
        self.log.info(text)
        sql9 = f'create table public.{self.table2}(column_a text);' \
               f'select * from public.{self.table1};'
        res9 = self.pri_sh.execut_db_sql(sql9,
                        sql_type=f'-U {self.user3} -W {macro.COMMON_PASSWD}')
        self.log.info(res9)
        self.assertIn('permission denied for schema public', res9,
                      f'执行失败: {text}')
        self.assertIn(f'permission denied for relation {self.table1}', res9,
                      f'执行失败: {text}')

    def tearDown(self):
        text = '----清理环境----'
        self.log.info(text)
        sql_cle1 = f'drop table {self.table1} cascade;'
        res_cle1 = self.pri_sh.execut_db_sql(sql_cle1)
        self.log.info(res_cle1)
        sql_cle2 = f'drop user {self.user1}, {self.user2}, {self.user3}, ' \
                   f'{self.user4} cascade;'
        res_cle2 = self.pri_sh.execut_db_sql(sql_cle2)
        self.log.info(res_cle2)
        self.assertEqual(self.constant.DROP_TABLE_SUCCESS, res_cle1,
                         f'执行失败: {text}')
        self.assertEqual(self.constant.DROP_ROLE_SUCCESS_MSG, res_cle2,
                         f'执行失败: {text}')
        self.log.info(f'-----{os.path.basename(__file__)} end-----')
